import asyncio
import errno
import json
import logging
import os
import pickle
import tarfile
import tempfile
import typing
import warnings
import zipfile
import glob
from asyncio import AbstractEventLoop
from io import BytesIO as IOReader
from pathlib import Path
from typing import Text, Any, Dict, Union, List, Type, Callable

import ruamel.yaml as yaml

from rasa.constants import ENV_LOG_LEVEL, DEFAULT_LOG_LEVEL

if typing.TYPE_CHECKING:
    from prompt_toolkit.validation import Validator

DEFAULT_ENCODING = "utf-8"


def configure_colored_logging(loglevel: Text) -> None:
    """配置log的颜色"""

    import coloredlogs

    loglevel = loglevel or os.environ.get(ENV_LOG_LEVEL, DEFAULT_LOG_LEVEL)

    field_styles = coloredlogs.DEFAULT_FIELD_STYLES.copy()
    field_styles["asctime"] = {}
    level_styles = coloredlogs.DEFAULT_LEVEL_STYLES.copy()
    level_styles["debug"] = {}
    coloredlogs.install(
        level=loglevel,
        use_chroot=False,
        fmt="%(asctime)s %(levelname)-8s %(name)s  - %(message)s",
        level_styles=level_styles,
        field_styles=field_styles,
    )


def enable_async_loop_debugging(
    event_loop: AbstractEventLoop, slow_callback_duration: float = 0.1
) -> AbstractEventLoop:
    """启用协程调试"""

    logging.info(
        "Enabling coroutine debugging. Loop id {}.".format(id(asyncio.get_event_loop()))
    )

    # 启用调试
    event_loop.set_debug(True)

    # 为了说明起见，使“slow”任务的阈值非常小。 默认值为0.1（= 100毫秒）
    event_loop.slow_callback_duration = slow_callback_duration

    # 报告所有管理异步资源的错误
    warnings.simplefilter("always", ResourceWarning)
    return event_loop


def fix_yaml_loader() -> None:
    """确保yaml读取的任何字符串都表示为unicode，避免编码格式报错"""

    def construct_yaml_str(self, node):
        # 覆盖默认的字符串处理函数以始终返回unicode对象
        return self.construct_scalar(node)

    yaml.Loader.add_constructor("tag:yaml.org,2002:str", construct_yaml_str)
    yaml.SafeLoader.add_constructor("tag:yaml.org,2002:str", construct_yaml_str)


def replace_environment_variables() -> None:
    """启用yaml加载器以处理yaml中的环境变量"""

    import re
    import os

    # eg. ${USER_NAME}, ${PASSWORD}
    env_var_pattern = re.compile(r"^(.*)\$\{(.*)\}(.*)$")
    yaml.add_implicit_resolver("!env_var", env_var_pattern)

    def env_var_constructor(loader, node):
        """处理yaml文件中的环境变量."""
        value = loader.construct_scalar(node)
        expanded_vars = os.path.expandvars(value)
        if "$" in expanded_vars:
            not_expanded = [w for w in expanded_vars.split() if "$" in w]
            raise ValueError(
                "Error when trying to expand the environment variables"
                " in '{}'. Please make sure to also set these environment"
                " variables: '{}'.".format(value, not_expanded)
            )
        return expanded_vars

    yaml.SafeConstructor.add_constructor("!env_var", env_var_constructor)


def read_yaml(content: Text) -> Union[List[Any], Dict[Text, Any]]:
    """从content中解析yaml
     Args:
        content: 包含Yaml内容的文本
    """

    fix_yaml_loader()                   # 修复yaml编码格式加载
    replace_environment_variables()     # 处理yaml文件中的环境变量

    yaml_parser = yaml.YAML(typ="safe")
    yaml_parser.version = "1.2"
    yaml_parser.unicode_supplementary = True

    # noinspection PyUnresolvedReferences
    try:
        return yaml_parser.load(content) or {}
    except yaml.scanner.ScannerError:
        # A `ruamel.yaml.scanner.ScannerError` might happen due to escaped
        # unicode sequences that form surrogate pairs. Try converting the input
        # to a parsable format based on（`ruamel.yaml.scanner.ScannerError`
        # 可能会由于形成代理对的转义unicode序列而发生。尝试根据以下内容将输入转换为可解析的格式）
        # https://stackoverflow.com/a/52187065/3429596.
        content = (
            content.encode("utf-8")
            .decode("raw_unicode_escape")
            .encode("utf-16", "surrogatepass")
            .decode("utf-16")
        )
        return yaml_parser.load(content) or {}


def read_file(filename: Text, encoding: Text = DEFAULT_ENCODING) -> Any:
    """读文件"""
    try:
        with open(filename, encoding=encoding) as f:
            return f.read()
    except FileNotFoundError:
        raise ValueError(f"File '{filename}' does not exist.")


def read_json_file(filename: Text) -> Any:
    """从文件中读取内容，并load成json格式"""
    content = read_file(filename)
    try:
        return json.loads(content)
    except ValueError as e:
        raise ValueError(
            "Failed to read json from '{}'. Error: "
            "{}".format(os.path.abspath(filename), e)
        )


def dump_obj_as_json_to_file(filename: Text, obj: Any) -> None:
    """将对象作为json字符串转储到文件中。"""
    write_text_file(json.dumps(obj, indent=2), filename)


def pickle_dump(filename: Union[Text, Path], obj: Any) -> None:
    """将对象保存到文件。
    Args:
        filename:   将对象保存到的文件名
        obj:        要存储的对象
    """
    with open(filename, "wb") as f:
        pickle.dump(obj, f)


def pickle_load(filename: Union[Text, Path]) -> Any:
    """Loads an object from a file.

    Args:
        filename: the filename to load the object from

    Returns: the loaded object
    """
    with open(filename, "rb") as f:
        return pickle.load(f)


def read_config_file(filename: Text) -> Dict[Text, Any]:
    """解析yaml配置文件，内容必须是字典
     参数:
        filename: 应当读取的文件的路径
    """
    content = read_yaml(read_file(filename))   # 读取端点内容

    if content is None:
        return {}
    elif isinstance(content, dict):
        return content
    else:
        raise ValueError(
            "Tried to load invalid config file '{}'. "
            "Expected a key value mapping but found {}"
            ".".format(filename, type(content))
        )


def read_yaml_file(filename: Text) -> Union[List[Any], Dict[Text, Any]]:
    """Parses a yaml file.

     Args:
        filename: The path to the file which should be read.
    """
    return read_yaml(read_file(filename, DEFAULT_ENCODING))


def unarchive(byte_array: bytes, directory: Text) -> Text:
    """Tries to unpack a byte array interpreting it as an archive.

    Tries to use tar first to unpack, if that fails, zip will be used."""

    try:
        tar = tarfile.open(fileobj=IOReader(byte_array))
        tar.extractall(directory)
        tar.close()
        return directory
    except tarfile.TarError:
        zip_ref = zipfile.ZipFile(IOReader(byte_array))
        zip_ref.extractall(directory)
        zip_ref.close()
        return directory


def write_yaml_file(data: Dict, filename: Union[Text, Path]) -> None:
    """写入yaml文件。
     Args:
        data: 要写入的数据。
        filename: 应写入文件的路径。
    """
    with open(str(filename), "w", encoding=DEFAULT_ENCODING) as outfile:
        yaml.dump(data, outfile, default_flow_style=False, allow_unicode=True)


def write_text_file(
    content: Text,
    file_path: Union[Text, Path],
    encoding: Text = DEFAULT_ENCODING,
    append: bool = False,
) -> None:
    """将文本写入文件。
    Args:
        content: 要写入的内容。
        file_path: 内容应写入的路径。
        encoding: 应该使用的编码。
        append: 是追加到文件还是截断文件。
    """
    mode = "a" if append else "w"
    with open(file_path, mode, encoding=encoding) as file:
        file.write(content)


def is_subdirectory(path: Text, potential_parent_directory: Text) -> bool:
    if path is None or potential_parent_directory is None:
        return False

    path = os.path.abspath(path)
    potential_parent_directory = os.path.abspath(potential_parent_directory)

    return potential_parent_directory in path


def create_temporary_file(data: Any, suffix: Text = "", mode: Text = "w+") -> Text:
    """为数据创建一个tempfile.NamedTemporaryFile对象。

    mode在py3中定义NamedTemporaryFile的mode参数。"""

    encoding = None if "b" in mode else DEFAULT_ENCODING
    f = tempfile.NamedTemporaryFile(
        mode=mode, suffix=suffix, delete=False, encoding=encoding
    )
    f.write(data)

    f.close()
    return f.name


def create_temporary_directory() -> Text:
    """Creates a tempfile.TemporaryDirectory."""
    f = tempfile.TemporaryDirectory()
    return f.name


def create_path(file_path: Text) -> None:
    """Makes sure all directories in the 'file_path' exists."""

    parent_dir = os.path.dirname(os.path.abspath(file_path))
    if not os.path.exists(parent_dir):
        os.makedirs(parent_dir)


def create_directory_for_file(file_path: Union[Text, Path]) -> None:
    """Creates any missing parent directories of this file path."""

    create_directory(os.path.dirname(file_path))


def file_type_validator(
    valid_file_types: List[Text], error_message: Text
) -> Type["Validator"]:
    """Creates a `Validator` class which can be used with `questionary` to validate
       file paths.
    """

    def is_valid(path: Text) -> bool:
        return path is not None and any(
            [path.endswith(file_type) for file_type in valid_file_types]
        )

    return create_validator(is_valid, error_message)


def not_empty_validator(error_message: Text) -> Type["Validator"]:
    """Creates a `Validator` class which can be used with `questionary` to validate
    that the user entered something other than whitespace.
    """

    def is_valid(input: Text) -> bool:
        return input is not None and input.strip() != ""

    return create_validator(is_valid, error_message)


def create_validator(
    function: Callable[[Text], bool], error_message: Text
) -> Type["Validator"]:
    """Helper method to create `Validator` classes from callable functions. Should be
    removed when questionary supports `Validator` objects."""

    from prompt_toolkit.validation import Validator, ValidationError
    from prompt_toolkit.document import Document

    class FunctionValidator(Validator):
        @staticmethod
        def validate(document: Document) -> None:
            is_valid = function(document.text)
            if not is_valid:
                raise ValidationError(message=error_message)

    return FunctionValidator


def list_files(path: Text) -> List[Text]:
    """返回除隐藏文件以外的所有文件。

     如果路径指向文件，则返回该文件"""

    return [fn for fn in list_directory(path) if os.path.isfile(fn)]


def list_subdirectories(path: Text) -> List[Text]:
    """Returns all folders excluding hidden files.

    If the path points to a file, returns an empty list."""

    return [fn for fn in glob.glob(os.path.join(path, "*")) if os.path.isdir(fn)]


def _filename_without_prefix(file: Text) -> Text:
    """Splits of a filenames prefix until after the first ``_``."""
    return "_".join(file.split("_")[1:])


def list_directory(path: Text) -> List[Text]:
    """返回除隐藏文件之外的所有文件和文件夹。

    如果路径指向文件，则返回该文件。这是一个递归实现，它返回路径任何深度的文件。"""

    if not isinstance(path, str):
        raise ValueError(
            "`resource_name` must be a string type. "
            "Got `{}` instead".format(type(path))
        )

    # 单个文件则直接返回
    if os.path.isfile(path):
        return [path]

    # 如果是文件夹，则递归查询
    elif os.path.isdir(path):
        results = []
        for base, dirs, files in os.walk(path, followlinks=True):
            # 在运行中以相同顺序对文件排序
            files = sorted(files, key=_filename_without_prefix)
            # 添加非隐藏文件
            good_files = filter(lambda x: not x.startswith("."), files)
            results.extend(os.path.join(base, f) for f in good_files)
            # 添加非隐藏目录
            good_directories = filter(lambda x: not x.startswith("."), dirs)
            results.extend(os.path.join(base, f) for f in good_directories)
        return results
    else:
        raise ValueError(
            "Could not locate the resource '{}'.".format(os.path.abspath(path))
        )


def create_directory(directory_path: Text) -> None:
    """创建目录及其父（super）路径。

     即使该路径已存在也同样会创建成功。"""

    try:
        os.makedirs(directory_path)
    except OSError as e:
        # 如果有人已经创建了路径，请感到高兴
        if e.errno != errno.EEXIST:
            raise


def zip_folder(folder: Text) -> Text:
    """从文件夹封存文件"""
    import tempfile
    import shutil

    zipped_path = tempfile.NamedTemporaryFile(delete=False)
    zipped_path.close()

    # WARN: not thread-safe!（不是线程安全的！）
    return shutil.make_archive(zipped_path.name, "zip", folder)


def json_unpickle(file_name: Union[Text, Path]) -> Any:
    """使用json从文件中解析对象。

    Args:
        file_name: 要从中加载对象的文件
    Returns: 对象
    """
    import jsonpickle.ext.numpy as jsonpickle_numpy
    import jsonpickle

    jsonpickle_numpy.register_handlers()

    file_content = read_file(file_name)
    return jsonpickle.loads(file_content)


def json_pickle(file_name: Union[Text, Path], obj: Any) -> None:
    """使用json序列化对象到文件（Pickle an object to a file using json）。

    Args:
        file_name: 将要保存对象的文件（the file to store the object to）
        obj: 要保存的对象
    """
    import jsonpickle.ext.numpy as jsonpickle_numpy
    import jsonpickle

    jsonpickle_numpy.register_handlers()

    write_text_file(jsonpickle.dumps(obj), file_name)
